Description

#16709

Additional context and related issues

This pr is going to support MERGE for jdbc based connectors, follows the #16693. Here are some major changes made in this pr:
JdbcClient & BaseJdbcClient:

1. Add getPrimaryKeys method to get primary keys for the merge target table, the method must not return empty if the connector declares supporting merge. The default implementation in BaseJdbcClient, and extracted function extractJdbcHandlesFromResultSet from getColumns for reuse it. Currently we use primary keys for perform delete and update for the merge.

2. Using return result of getPrimaryKeys to check if the connector support merge.

3. Add methods beginDeleteTableForMerge and finishDeleteTableForMerge for supporting FTE about the merge, the logic will be used if the connector supports retry and use transactional insert. The default implementation in BaseJdbcClient :

   1. beginDeleteTableForMerge method uses primary keys to build a temporary table

   2. finishDeleteTableForMerge uses syntax DELETE FROM merge_target WHERE EXISTS (SELECT 1 FROM (temp_table_data ) temp WHERE "getConjunctsBetweenTargetAndTemporaryTable(merge_target, temp)" ) to perform actual delete.

      temp_table_data is the sth like SELECT * FROM temp_table WHERE EXISTS (SELECT 1 FROM page_sink_table WHERE page_sink_table.id_column = temp_table.id_column) The logic is the same as original INSERT operation that using temporary table(INSERT INTO insert_target SELECT * FROM temp_table WHERE EXISTS (SELECT 1 FROM page_sink_table WHERE page_sink_table.id_column = temp_table.id_column)), the temp_table is built in beginDeleteTableForMerge

      The connectors returns special primary keys that can not build temporary table with get primary keys directly , i.e PostgreSql returns ctid as the primary key, it is a hidden column for every table, while building temporary table storing the column data we need to rename the special column for avoiding conflicts, in PostgreSql we build the temporary table with column name ctid_for_delete_merge to store the ctid values, then use getConjunctsBetweenTargetAndTemporaryTable to use correct column name to build the condition between target and temp table .

4. Add method buildMergeRowIdConjuncts for the connectors that not support(or disable) building temporary table to perform delete, the method builds the WHERE condition for clause DELETE FROM merge_target WHERE... using primary keys (called in JdbcMergeSink )

5. The method updateScanColumnsForMerge is the same as we support merge in Phoenix, it updates the scans for including the all the primary keys if possible.

DefaultJdbcMetadata:

1. Add a deleteForMergeRollbackAction field for the rollback about the delete operation if the connectors supports delete using temporary table.
2. The method getMergeRowIdColumnHandle will check if the connector support merge, if yes it will return the column handle that composed by the primary keys(returned by getPrimaryKeys in client)
3. beginMerge will pass the all the infos that needed by the merge, it's similar to we do in Phoenix connector, the difference is we add a beginDeleteMerge process for connectors that will use temporary table to perform delete.
4. finishMerge calls the finishInsert and finishDeleteForMerge

JdbcMergeSink:
It's moved from the Phoenix implementation, difference are:

• the creating delete sink will return the insert sink with the temporary table handle if the connectors support delete using temporary table.
• The update will use syntax UPDATE merge_target SET reserved_columns WHERE..., the reserved_columns are the columns who are not the primary keys, the values come from channels that provided by getReservedChannelsForUpdate , and the WHERE condition is built by jdbcClient.buildMergeRowIdConjuncts.

The connectors:
Implementation about the getPrimaryKeys :

• PostgreSql using the ctid
• Oracle using the ROWID
• Ignite using the primary keys defined in table
• Phoenix using the rowkey defined in table
  Other Jdbc connectors Like Mysql actual can perform the merge as long as the table has defined the primary key, but since it is not forced by the connector, we currently let the BaseJdbcClient.supportsMerge return false by default and not declare it supports merge.

Tests:
PostgreSql and Oracle support FTE, so the defined the supportMerge in the BaseJdbcFailureRecoveryTest for indicating the behavior, aslo update the test cases through the method in BaseJdbcFailureRecoveryTest.
Adjust lots of merge tests in Oracle due to naming length constraints.

Release notes

( ) This is not user-visible or docs only and no release notes are required.
(x) Release notes are required, please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Base Jdbc
* Support SQL MERGE for base jdbc connectors ({issue}`16709`)
* Support update/merge for PostgreSql
* Support update/merge for Ignite
* Support update/merge for Oracle
* Update Phoenix merge implementation by reusing base Jdbc merge implementation.